/*
    Copyright 2010 Northbranchlogic, Inc.

    Licensed under the Apache License, Version 2.0 (the "License");
    you may not use this file except in compliance with the License.
    You may obtain a copy of the License at

        http://www.apache.org/licenses/LICENSE-2.0

    Unless required by applicable law or agreed to in writing, software
    distributed under the License is distributed on an "AS IS" BASIS,
    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    See the License for the specific language governing permissions and
    limitations under the License.
 */

package com.norbl.runjrun;

import java.io.*;
import java.util.*;

/** Groups a list of tasks (<tt>Runnable</tt>, <tt>Serializable</tt> objects)
 *  into groups of tasks.  The <tt>run()</tt> method of <tt>ThreadedRunnable</tt>
 *  creates a thread for each group.  Each thread runs its subset of tasks
 *  sequentially.
 *
 * @author Barnet Wagman
 */
public class ThreadedRunnable implements Runnable, Serializable {

    static final long serialVersionUID = 0L;

    public static long NAP_TIME = 5L * 1000L;

    private int nThreads;
    private List<Runnable>[] groups;

    public List<Throwable> throwables;

        /**
         * 
         * @param runnables list of tasks; these object must be <tt>Serializable</tt>
         *  as well as <tt>Runnable</tt>.
         * @param nThreads number of threads used to execute the tasks in
         *  the <tt>runnables</tt> list.
         */
    public ThreadedRunnable(List<Runnable> runnables, int nThreads) {
        if ( nThreads < 1 )
            throw new RunJRunException("nThreads=" +
                        nThreads + "; it must be >=1");
        this.nThreads = nThreads;
        groups = groupRunnables(runnables,nThreads);
    }

        /** Blocks until all threads are done.
         */
    public void run() {
        try {
            throwables = new ArrayList<Throwable>();
            List<Runner> runners = new ArrayList<Runner>();

            for ( List<Runnable> g : groups ) {
                Runner r = new Runner(g);
                Thread t = new Thread(r);
                t.start();
                runners.add(r);
            }
            waitForThreadsToFinish(runners);

            for ( Runner r : runners ) {
                List<Throwable> xs = r.throwables;
                if ( xs != null ) throwables.addAll(xs);
            }
        }
        catch(Throwable xxx) {
            throwables.add(xxx);
        }
    }

    public List<Throwable> getThrowables() { return(throwables); }

        /** Creates a list of {@link ThreadedRunnable}s. The <tt>runnables</tt>
         *  are divided into <tt>nTask</tt> groups. Each group of
         *  <tt>runnables</tt> is assigned to a separate instance of
         *  <tt>ThreadedRunnable</tt>.
         *
         * @param runnables list of tasks; these object must be <tt>Serializable</tt>
         *  as well as <tt>Runnable</tt>.
         * @param nTasks the number of <tt>ThreadedRunnable</tt>s that will be
         *  created.
         * @param nThreads number of threads used to execute the tasks in each
         *      <tt>ThreadedRunnable</tt> instances.
         * @return a <tt>List</tt> <tt>ThreadedRunnable</tt>s; note that
         *  the return is declared as <tt>List&lt;Runnable&gt;</tt> so
         *  that it can be passed to {@link Client#submit(java.util.List)}
         *  without repackaging.
         */
    public static List<Runnable> create(List<Runnable> runnables,
                                        int nTasks,
                                        int nThreads) {
            // Divide the runables into nThreadedRunnable list.
        List<Runnable>[] nodeGroups =
                groupRunnables(runnables, nTasks);

           // Create the ThreadedRunnables
        List<Runnable> trs = new ArrayList<Runnable>();
        for ( List<Runnable> rs : nodeGroups ) {
            trs.add(new ThreadedRunnable(rs,nThreads));
        }
        return(trs);
    }

        /** Retrieves <tt>ThreadedRunnable</tt>s and unpacks the
         *  tasks within them. This methods blocks until all tasks
         *  are returned.  <i>Warning</i>: This method assumes that
         *  the <tt>TaskDescription</tt>s <tt>des</tt> all describe
         *  <tt>ThreadedRunnable</tt>.  If that isn't true, you'll
         *  get a <tt>ClassCastException</tt>.
         *
         * @param client
         * @param des
         * @return cooked tasks.
         */
    public static List<Runnable> retrieve(Client client,
                                          List<TaskDescription> des) {

        List<Runnable> trs = client.getCooked(des);
        List<Runnable> tasks = new ArrayList<Runnable>();

        for ( Runnable tr : trs ) {
            tasks.addAll( ((ThreadedRunnable) tr).getRunnables() );
        }
        return(tasks);
    }
                                          

    public int getNGroups() {
        if ( groups != null ) return(groups.length);
        else return(0);
    }

    public List<Runnable> getRunnables() {

        if ( groups == null ) return(new ArrayList<Runnable>());

        List<Runnable> rs = new ArrayList<Runnable>();
        for ( List<Runnable> g : groups ) {
            rs.addAll(g);
        }
        return(rs);
    }

    public int getNRunnables() {
        return( getRunnables().size() );
    }


        // ----------------------------------------

    private static List<Runnable>[] groupRunnables(List<Runnable> runnables,
                                                   int nGroups) {

        List<Runnable>[] groups = new List [nGroups];

        for ( int i = 0; i < nGroups; i++ )
            groups[i] = new ArrayList<Runnable>();

        int i = 0;
        while ( i < runnables.size() ) {
            for (int j = 0; j < groups.length; j++ ) {
                groups[j].add(runnables.get(i));
                ++i;
                if ( i >= runnables.size() ) {
                    checkCount(runnables,groups);
                    return(groups);
                }
            }
        }

        checkCount(runnables,groups);
        return(groups);
    }

    private static void checkCount(List<Runnable> runnables,
                                   List<Runnable>[] groups) {
            // Check the count
        int n = 0;
        for ( List<Runnable> g : groups ) {
            n += g.size();
        }
        if ( n != runnables.size() )
            throw new RunJRunException("BUG: n in groups=" + n +
                    " but it SHOULD=" + runnables.size());
    }

    class Runner implements Runnable, Serializable {
        static final long serialVersionUID = 0L;

        List<Runnable> group;
        boolean isDone;
        List<Throwable> throwables;

        public Runner(List<Runnable> group) {
            this.group = group;
            isDone = false;
        }
        public void run() {
            throwables = new ArrayList<Throwable>();
            for ( Runnable r : group ) {
                try {
                    r.run();
                }
                catch(Throwable t) {
                    throwables.add(t);
                }
            }
            isDone = true;
        }
    }

    private void waitForThreadsToFinish(List<Runner> runners) {

        while ( nRunning(runners) > 0 ) {
            try {
                Thread.sleep(NAP_TIME);
            }
            catch(InterruptedException ix) {}
        }
    }

    private int nRunning(List<Runner> runners) {
        int n = 0;
        for ( Runner r : runners ) {
            if ( !r.isDone ) ++n;
        }
        return(n);
    }
}
